package com.itheima.rpc.client.request;

import com.itheima.rpc.cache.ServiceProviderCache;
import com.itheima.rpc.client.cluster.LoadBalanceStrategy;
import com.itheima.rpc.client.cluster.StrategyProvider;
import com.itheima.rpc.client.config.RpcClientConfiguration;
import com.itheima.rpc.data.RpcRequest;
import com.itheima.rpc.data.RpcResponse;
import com.itheima.rpc.netty.codec.FrameDecoder;
import com.itheima.rpc.netty.codec.FrameEncoder;
import com.itheima.rpc.netty.codec.RpcRequestEncoder;
import com.itheima.rpc.netty.codec.RpcResponseDecoder;
import com.itheima.rpc.netty.handler.RpcResponseHandler;
import com.itheima.rpc.netty.request.ChannelMapping;
import com.itheima.rpc.netty.request.RequestPromise;
import com.itheima.rpc.netty.request.RpcRequestHolder;
import com.itheima.rpc.provider.ServiceProvider;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.FutureListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class RpcRequestManager {

    @Resource
    private ServiceProviderCache providerCache;

    @Resource
    private RpcClientConfiguration configuration;

    @Resource
    private StrategyProvider provider;

    public RpcResponse sendRequest(RpcRequest request) {
        //获取可用的服务提供者列表
        List<ServiceProvider> providerList = providerCache.get(request.getClassName());
        if (ObjectUtils.isEmpty(providerList)) {
            log.info("没有可用的提供者列表，className: {}", request.getClassName());
            return new RpcResponse();
        }
        LoadBalanceStrategy strategy = provider.getStrategy();
        return requestByNetty(request, strategy.select(providerList));
    }

    /**
     * 启动消费者netty
     * @param request
     * @param provider
     * @return
     */
    private RpcResponse requestByNetty(RpcRequest request, ServiceProvider provider) {
        if (!RpcRequestHolder.channelExist(provider.getServerIp(), provider.getRpcPort())) {
            EventLoopGroup group = new NioEventLoopGroup(0, new DefaultThreadFactory("client"));
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast("FrameEncoder", new FrameEncoder());
                            pipeline.addLast("RpcRequestEncoder", new RpcRequestEncoder());

                            pipeline.addLast("FrameDecoder", new FrameDecoder());
                            pipeline.addLast("RpcResponseDecoder", new RpcResponseDecoder());
                            pipeline.addLast("RpcResponseHandler", new RpcResponseHandler());
                        }
                    });
            try {
                ChannelFuture future = bootstrap.connect(provider.getServerIp(), provider.getRpcPort()).sync();
                if (!future.isSuccess()) {
                    log.info("连接服务端失败，className: {}", request.getClassName());
                    return new RpcResponse();
                }
                //channel复用  同一个提供者只创建一个客户端连接通道
                RpcRequestHolder.addChannelMapping(new ChannelMapping(provider.getServerIp(), provider.getRpcPort(), future.channel()));
            } catch (InterruptedException e) {
                log.error("客户端连接创建失败==============");
                group.shutdownGracefully();
                throw new RuntimeException(e);
            }
        }

        try {
            Channel channel = RpcRequestHolder.getChannel(provider.getServerIp(), provider.getRpcPort());
            //创建一个监听线程
            RequestPromise promise = new RequestPromise(channel.eventLoop());
            RpcRequestHolder.addRequestPromise(request.getRequestId(), promise);

            //发送数据
            ChannelFuture channelFuture = channel.writeAndFlush(request);
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (!channelFuture.isSuccess()) {
                        RpcRequestHolder.removeRequestPromise(request.getRequestId());
                    } else {
                        log.info("远程调用成功================");
                    }
                }
            });

            //监听回调
            promise.addListener(new FutureListener<RpcResponse>() {
                @Override
                public void operationComplete(Future<RpcResponse> responseFuture) throws Exception {
                    if (!responseFuture.isSuccess()) {
                        RpcRequestHolder.removeRequestPromise(request.getRequestId());
                    } else {
                        log.info("监听调用结果回调成功==============");
                    }
                }
            });

            //异步获取结果
            log.info("异步获取结果等待===============");
            RpcResponse response = (RpcResponse)promise.get(configuration.getConnectTimeout(), TimeUnit.SECONDS);
            log.info("远程调用结束，response={}", response);
            RpcRequestHolder.removeRequestPromise(request.getRequestId());
            return response;
        } catch (Exception e) {
            log.error("调用服务端失败,失败原因: {}", e.getMessage());
            throw new RuntimeException(e);
        }
    }
}
